package org.momo.rpc.protocol.codec;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import lombok.extern.slf4j.Slf4j;
import org.momo.rpc.constant.SerializeKey;
import org.momo.rpc.protocol.RpcMessage;
import org.momo.rpc.protocol.RpcRequest;
import org.momo.rpc.protocol.RpcResponse;
import org.momo.rpc.protocol.constant.MessageType;
import org.momo.rpc.protocol.constant.ProtocolConstant;
import org.momo.rpc.serialize.Serializer;
import org.momo.rpc.extention.ExtensionLoader;

import java.util.List;

@Slf4j
public class NettyDecodeHandler extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> list) throws Exception {
        // 判断当前buf中的可读数据长度是否大于Header，防止数据未到达
        if(byteBuf.readableBytes() < ProtocolConstant.HEADER_LENGTH) {
            return;
        }
        byteBuf.markReaderIndex();

        // magic
        byte magic = byteBuf.readByte();
        if(magic != ProtocolConstant.MAGIC) {
            throw new RuntimeException("Invalid MAGIC");
        }

        // 读取Header
        RpcMessage.Header header = new RpcMessage.Header();
        header.setMagic(magic);
        header.setVersion(byteBuf.readByte());
        header.setSerializer(byteBuf.readByte());
        header.setMessageType(byteBuf.readByte());
        header.setStatus(byteBuf.readByte());
        header.setId(byteBuf.readLong());
        header.setLength(byteBuf.readInt());

        // 获取 messageType
        MessageType messageType = MessageType.getMessageTypeEnumByCode(header.getMessageType());
        if(messageType == null) {
            log.error("Invalid message type");
            return;
        }

        // ping
        if(header.getMessageType() == MessageType.HEARTBEAT_PING.getCode()) {
            return;
        }

        // request, data长度数据
        if(byteBuf.readableBytes() < header.getLength()) {
            byteBuf.resetReaderIndex();
            return;
        }

        // 读取data
        byte[] data = new byte[header.getLength()];
        byteBuf.readBytes(data);

        // get serializer
        String serializerKey = SerializeKey.getSerializerKeyByCode(header.getSerializer());
        Serializer serializer = ExtensionLoader.getExtensionLoader(Serializer.class).getExtension(serializerKey);

        // 反序列化 data
        Object obj = null;
        switch (messageType) {
            case REQUEST:
                obj = serializer.deserialize(data, RpcRequest.class);
                break;
            case RESPONSE:
                obj = serializer.deserialize(data, RpcResponse.class);
                break;
            default:
                return;
        }

        RpcMessage<?> message = new RpcMessage<>(header, obj);
        list.add(message);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("Decode error! the error is: {}", cause.getMessage());
        super.exceptionCaught(ctx, cause);
    }

}
